package com.shengzai.stream

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo4Window2 {
  def main(args: Array[String]): Unit = {

    //读取前15秒的数据
    val conf = new SparkConf()
    conf.setMaster("local[2]")
    conf.setAppName("window")

    val sc = new SparkContext(conf)

    val ssc = new StreamingContext(sc, Durations.seconds(5))

    ssc.checkpoint("data/checkpoint")

    val ds: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.146.136", 8888)

    val mapDS: DStream[(String, Int)] = ds.flatMap(line => line.split(","))
      .map(line => (line, 1))

    val winRes: DStream[(String, Int)] = mapDS.reduceByKeyAndWindow(
      (x, y) => x + y,
      (x, y) => x - y,
      //取前15秒的数据
      windowDuration = Durations.seconds(15),
      //5秒钟滑动一次
      slideDuration = Durations.seconds(5)
    )

    println(winRes)

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
